﻿#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>

#include <memory>
#include <string>
#include <iostream>
#include <functional>
#include <thread>
#include <memory>
#include <mutex>


using boost::asio::ip::tcp;

// 服务器和某个客户端之间的“会话”
// 负责处理读写事件
class Session : public std::enable_shared_from_this<Session> {
public:
    explicit Session(tcp::socket &&s) : socket_{std::move(s)} {
        data_.fill('\0');
    }

    ~Session() {
//        std::cout << "[" << std::this_thread::get_id() << "] "
//                  << "Close Client["
//                  << socket_.remote_endpoint()
//                  << "] \n";
    }

    void start() {
        std::cout << "[" << std::this_thread::get_id() << "] "
                  << "Accept New Client["
                  << socket_.remote_endpoint()
                  << "] \n";
        async_read();
    }

    void stop() {
        socket_.close();
    }

private:
    void async_read() {
        // 异步读
        //socket_.async_receive()
        socket_.async_read_some(
                boost::asio::buffer(data_),  // 数据buffer
                [this, self{shared_from_this()}](
                        const boost::system::error_code &ec,
                        size_t bytes_transferred) {
                    if (!ec) {
                        std::scoped_lock locker{putMtx_};
                        std::cout << "[" << std::this_thread::get_id() << "] "
                                  << "Client["
                                  << socket_.remote_endpoint()
                                  << "] <<< "
                                  << std::string{data_.data(), bytes_transferred}
                                  << "\n";
                        self->async_write(bytes_transferred);  // 读完即写
                    } else {
                        socket_.close();
                    }
                });
    }

    void async_write(std::size_t length) {
        // 异步写
        socket_.async_write_some(
                boost::asio::buffer(data_, length),
                [this, self{shared_from_this()}](
                        const boost::system::error_code &ec,
                        size_t bytes_transferred) {
                    if (!ec) {
                        std::scoped_lock locker{putMtx_};
                        std::cout << "[" << std::this_thread::get_id() << "] "
                                  << "Client["
                                  << socket_.remote_endpoint()
                                  << "] >>> "
                                  << std::string{data_.data(), bytes_transferred}
                                  << "\n";
                        async_read();
                    } else {
                        socket_.close();
                    }
                });
        /*
        boost::asio::async_write(
                socket_,
                boost::asio::buffer(data_, length),  // 异步写
                [this, self{shared_from_this()}](const boost::system::error_code &ec, size_t) {
                    if (!ec)
                        async_read();
                }
        );
         */
    }

    std::mutex putMtx_;
    tcp::socket socket_;  // “会话”基于已经建立的socket连接
    std::array<char, 1024> data_;
};

// 服务器类
// 监听客户端连接请求（async_accept）。与某个客户端建立socket连接后，为它创建一个session
class Server {
public:
    Server(boost::asio::io_context &io_ctx, short port)
            : acceptor_(io_ctx, tcp::endpoint(tcp::v4(), port)),
              socket_(io_ctx) {
        async_accept();
    }

private:
    void async_accept() {
        // 异步accept。socket连接建立后，调用handle_accept()
        acceptor_.async_accept(socket_, [this](auto &&PH1) {
            handle_accept(std::forward<decltype(PH1)>(PH1));
        });
    }

    void handle_accept(const boost::system::error_code &ec) {
        if (!ec) {
            //被移动后并不一定是将亡值，还可以作为任何用途使用。所以必须是一个合法状态。
            auto pSession = std::make_shared<Session>(std::move(socket_));
            pSession->start();
        }
        async_accept();  // 继续监听客户端连接请求
    }

    tcp::acceptor acceptor_; // 接收器
    tcp::socket socket_;     // socket对象
};

int main(int argc, char *argv[]) {
    if (argc != 3) {
        std::cerr << "<port> <threads>\n";
        return 1;
    }
    auto port = boost::lexical_cast<short>(argv[1]);
    auto nThreads = boost::lexical_cast<int>(argv[2]);
    try {
        std::vector<std::thread> threads;
        boost::asio::io_context io_ctx;

        for (int i = 0; i < nThreads; ++i) {
            threads.emplace_back([&]() {
                io_ctx.run();
            });
        }

        Server svr(io_ctx, port);
        Server svr2(io_ctx, port + 1);

        std::string line;
        while (true) {
            std::getline(std::cin, line);
            if (line == "exit") {
                io_ctx.stop();
                break;
            }
        }

        for (auto &t: threads) {
            t.join();
        }
    }
    catch (std::exception &e) {
        std::cerr << "Exception: " << e.what() << "\n";
    }
    return 0;
}